package com.yonyou.cn.mongodbtest.utils;

import com.alibaba.fastjson.JSON;
import com.mongodb.*;
import com.mongodb.client.FindIterable;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoCursor;
import com.mongodb.client.MongoDatabase;
import com.mongodb.client.model.Filters;

import org.apache.commons.lang3.StringUtils;
import com.yonyou.cn.mongodbtest.constant.MongodbConst;
import org.apache.commons.collections4.MapUtils;
import org.bson.Document;
import org.bson.conversions.Bson;
import org.bson.types.Decimal128;

//import org.springframework.util.Assert;
import org.apache.log4j.Logger;

import java.util.*;

public class MongoDBUtils {
	public static Logger log = Logger.getLogger(MongoDBUtils.class);

	private static final Integer FIRST_PAGE_NUM = 1;
	
	private static MongoClient mongoClient = null;  
	
	/*private final static String ip = "localhost"; //ip
	private final static int port = 27017; //端口
	private final static String username = "yhs"; //用户名
	private final static String password = "123456"; //密码
	private final static String dbname = "admin"; //库名--保存数据用
	 */


	/**
	 * @author yhs
	 * @function 获取mongodb连接对象
	 * ip       服务器地址
	 * port     端口号，默认27017
	 * username 用户名
	 * password 密码
	 * dbname   库名
	 * table    表名
	 */
	public static MongoCollection<Document> getCollection(String ip, int port, String username, String password, String dbname, String table) {

		MongoCollection<Document> collection = null;
		try {
			MongoClientOptions.Builder build = new MongoClientOptions.Builder();
			build.connectionsPerHost(300);  //连接池设置为300个连接,默认为100
			build.threadsAllowedToBlockForConnectionMultiplier(5000); //线程队列数，如果连接线程排满了队列就会抛出“Out of semaphores to get db”错误
			build.connectTimeout(60000); // 连接超时，推荐>3000毫秒
			build.maxWaitTime(10000); //从连接池中获取Collection的最大等待时间
			build.socketTimeout(0);// 套接字超时时间，0无限制
			MongoClientOptions options = build.build();

			List<ServerAddress> addrs = new ArrayList<ServerAddress>();
			ServerAddress serverAddress = new ServerAddress(ip, port);
			addrs.add(serverAddress);

			MongoCredential credential = MongoCredential.createScramSha1Credential(username, dbname, password.toCharArray());
			List<MongoCredential> credentials = new ArrayList<MongoCredential>();
			credentials.add(credential);
			MongoClient mongoClient = new MongoClient(addrs, credentials, options);
			//mongoClient = new MongoClient(addrs, credentials, options);
			System.out.println("mongodb服务连接成功");
			//连接到数据库
			MongoDatabase mongoDatabase = mongoClient.getDatabase(dbname);//选择数据库,如果mongoDB中没有这个数据库,当向此库中添加数据的时候会自动创建
			System.out.println("数据库选择成功");
			collection = mongoDatabase.getCollection(table);//选择表,如果mongoDB中没有这个表,当向此表中添加数据的时候会自动创建
			System.out.println("表选择成功");
		} catch (Exception e) {
			e.printStackTrace();
		}
		return collection;
	}

	public static MongoCollection<Document> getDBConnection(String dbname, String table, List<String> ips, int port,String username, String password) {
		   MongoCollection<Document> collection = null;
			if(mongoClient == null){  
	            intializeMongoClient(ips,port,username,password,dbname);           
	        }
			System.out.println("mongodb服务连接成功");
			//连接到数据库
			MongoDatabase mongoDatabase = mongoClient.getDatabase(dbname);//选择数据库,如果mongoDB中没有这个数据库,当向此库中添加数据的时候会自动创建
			System.out.println("数据库选择成功");
			collection = mongoDatabase.getCollection(table);//选择表,如果mongoDB中没有这个表,当向此表中添加数据的时候会自动创建
			System.out.println("表选择成功");
		    return collection;
	}

	public static void main(String[] args)  throws Exception{
//		for(int x=12;x<21;x++) {
			MongoCollection<Document> collection = null;
			if (mongoClient == null) {
				intializeMongoClient(Arrays.asList(new String[]{"10.10.65.6"}), 28017, "root", "root", "admin");
			}
			System.out.println("mongodb服务连接成功");
			//连接到数据库
			MongoDatabase mongoDatabase = mongoClient.getDatabase("admin");//选择数据库,如果mongoDB中没有这个数据库,当向此库中添加数据的时候会自动创建
			System.out.println("数据库选择成功");
//		collection = mongoDatabase.getCollection(table);//选择表,如果mongoDB中没有这个表,当向此表中添加数据的时候会自动创建
//		System.out.println("表选择成功");
//		return collection;


			String kubiao = "czsb.test09";


			BasicDBObject bs = new BasicDBObject();
			bs.append("shardCollection", kubiao);
			Map map = new HashMap();
			map.put("month", 1);
			bs.append("key", JSON.toJSON(map));
			System.out.println(bs.toJson());
			mongoDatabase.runCommand(bs);

			for (int i = 1; i <= 12; i++) {
				BasicDBObject bs1 = new BasicDBObject();
				bs1.append("split", kubiao);
				Map map1 = new HashMap();
				map1.put("month", "" + i);
				bs1.append("middle", JSON.toJSON(map1));
				System.out.println(bs1);
				mongoDatabase.runCommand(bs1);
			}
//		}

	}
	
	private static void intializeMongoClient( List<String> ips, int port,String username, String password,String dbname) {  
		MongoClientOptions.Builder build = new MongoClientOptions.Builder();
		build.connectionsPerHost(300);  //连接池设置为300个连接,默认为100
		build.threadsAllowedToBlockForConnectionMultiplier(5000); //线程队列数，如果连接线程排满了队列就会抛出“Out of semaphores to get db”错误
		build.connectTimeout(60000); // 连接超时，推荐>3000毫秒
		build.maxWaitTime(10000); //从连接池中获取Collection的最大等待时间
		//build.socketTimeout(0);// 套接字超时时间，0无限制
		build.socketKeepAlive(true);
		MongoClientOptions options = build.build();
		if(StringUtil.isEmpty(username) || StringUtil.isEmpty(password)){
			throw new RuntimeException("====mongodb用户名或密码配置有误，请火速联系管理员====");
		}
		MongoCredential credential = MongoCredential.createScramSha1Credential(username, dbname, password.toCharArray());
		List<MongoCredential> credentials = new ArrayList<MongoCredential>();
		credentials.add(credential);
		List<ServerAddress> addrs = new ArrayList<ServerAddress>();
		for (String ip : ips) {
			ServerAddress serverAddress = new ServerAddress(ip, port);
			addrs.add(serverAddress);
		}
        mongoClient = new MongoClient( addrs ,credentials, options );
         
   }  


	//保存
	public static void save(MongoCollection<Document> collection, List<Document> documents) {
		Document document = new Document("name", "guanxiaotong").append("age", 18);
		documents.add(document);
		collection.insertMany(documents);
	}

	//更新
	public static void update(MongoCollection<Document> collection, BasicDBObject query) {
		collection.updateMany(query, new Document("$set", new Document("age", 20)));//将年龄改为20
	}

	//查询
	public static void select(MongoCollection<Document> collection, BasicDBObject query, int pageNo, int pageSize) {
		FindIterable<Document> resultIterable = collection.find(query).skip((pageNo - 1) * pageSize).limit(pageSize);
		long count = collection.count(query); //查询满足条件的总数		
	}

	//删除
	public static void delete(MongoCollection<Document> collection, BasicDBObject query) {
		query.put("name", "jack");//姓名等于jack
		query.put("age", 20);//年龄等于20
		query.put("score", new BasicDBObject("$gt", 60).append("$lt", 80));//成绩大于60小于80
		collection.deleteMany(query);
	}

	/**
	 * @param collection
	 * @param queryInfo  查询条件
	 * @param sort       排序字段
	 * @return
	 * @Title: query
	 * @Description: 不分页查询
	 */
	public static List<Object> query(MongoCollection<Document> collection, Bson queryInfo, BasicDBObject sort) {
		List<Object> dataList = null;
		if (sort != null) {
			dataList = collection.find(queryInfo).sort(sort).into(new ArrayList<>());
		} else {
			dataList = collection.find(queryInfo).into(new ArrayList<>());
		}
		return dataList;
	}


	/**
	 * @param collection
	 * @param queryParam
	 * @return
	 * @Title: aggregate
	 * @Description: 聚合查询
	 */
	public static List<Object> aggregate(MongoCollection<Document> collection, Map<String, Object> queryParam) {
		List<Document> pipeline = packagePipline(queryParam);
		Document s = new Document("$skip",100);
		pipeline.add(s);
		Document l = new Document("$limit",100);
		pipeline.add(l);

		MongoCursor<Document> iterator = collection.aggregate(pipeline).allowDiskUse(true).iterator();

		List<Object> result = new ArrayList<>();
		while (iterator.hasNext()) {
			result.add(iterator.next());
		}
		return result;
	}
	public static List<Object> aggregatePage(MongoCollection<Document> collection, Map<String, Object> queryParam) {
//		long countstart = System.currentTimeMillis();
//		//查询总数
//		long total = count(collection,(Document)queryParam.get("match"));
//		long counend = System.currentTimeMillis();
//		System.out.println("统计共用时:" + ((counend - countstart) / 1000) + "秒");		long countstart = System.currentTimeMillis();
//		//查询总数
//		long total = count(collection,(Document)queryParam.get("match"));
//		long counend = System.currentTimeMillis();
//		System.out.println("统计共用时:" + ((counend - countstart) / 1000) + "秒");

//
//		Integer pageNo = (Integer) queryParam.get("pageNo");
//		Integer pageSize = (Integer) queryParam.get("pageSize");
//		final Integer pages = (int) Math.ceil(total / (double) pageSize);
//		if (pageNo <= 0) {
//			pageNo = FIRST_PAGE_NUM;
//		}
//		int skip = pageSize * (pageNo - 1);
//
//		List<Document> pipeline = packagePipline(queryParam);
//		Document s = new Document("$skip",100);
//		pipeline.add(s);
//		Document l = new Document("$limit",pages);
//		pipeline.add(l);


		long addstart = System.currentTimeMillis();
		MongoCursor<Document> iterator = collection.aggregate(null).allowDiskUse(true).iterator();
		long addend = System.currentTimeMillis();
		System.out.println("聚合共用时:" + ((addend - addstart) / 1000) + "秒");

		List<Object> result = new ArrayList<>();
		while (iterator.hasNext()) {
			result.add(iterator.next());
		}
		return result;
	}
	/**
	 * @param queryParam
	 * @return
	 * @Title: packagePipline
	 * @Description: 组装聚合条件
	 */
	private static List<Document> packagePipline(Map<String, Object> queryParam) {
		List<Document> result = new ArrayList<Document>();
		if (MapUtils.isEmpty(queryParam)) {
			return result;
		}
		queryParam.forEach((k, v) -> {
			if (StringUtils.equals(k, MongodbConst.AggregateOperEnum.OPER_MATCH.getKey())) {
				result.add(new Document(MongodbConst.AggregateOperEnum.OPER_MATCH.getValue(), v));
			} else if (StringUtils.equals(k, MongodbConst.AggregateOperEnum.OPER_GROUP.getKey())) {
				result.add(new Document(MongodbConst.AggregateOperEnum.OPER_GROUP.getValue(), v));
			} else if (StringUtils.equals(k, MongodbConst.AggregateOperEnum.OPER_PROJECT.getKey())) {
				result.add(new Document(MongodbConst.AggregateOperEnum.OPER_PROJECT.getValue(), v));
			} else if (StringUtils.equals(k, MongodbConst.AggregateOperEnum.OPER_SORT.getKey())) {
				result.add(new Document(MongodbConst.AggregateOperEnum.OPER_SORT.getValue(), v));
			} else if(StringUtils.equals(k, MongodbConst.AggregateOperEnum.OPER_GROUP_AGAIN.getKey())) {
				result.add(new Document(MongodbConst.AggregateOperEnum.OPER_GROUP_AGAIN.getValue(), v));
			}else if(StringUtils.equals(k, MongodbConst.AggregateOperEnum.OPER_LIMIT.getKey())) {
				result.add(new Document(MongodbConst.AggregateOperEnum.OPER_LIMIT.getValue(), v));
			}else if(StringUtils.equals(k, MongodbConst.AggregateOperEnum.OPER_SKIP.getKey())) {
				result.add(new Document(MongodbConst.AggregateOperEnum.OPER_SKIP.getValue(), v));
			}
			else {

			}
		});
		return result;
	}


	/**
	 * @param collection
	 * @param queryParam   查询参数 key为：query，sort，pageNo，pageSize
	 * @return
	 * @Title: pageQuery
	 * @Description: 分页查询指定数据
	 */
	public static Map<String, Object> pageQuery(MongoCollection<Document> collection, Map<String, Object> queryParam) {
		Document query = (Document) queryParam.get(MongodbConst.AggregateOperEnum.OPER_MATCH.getKey());
		Document sort = (Document) queryParam.get(MongodbConst.AggregateOperEnum.OPER_SORT.getKey());
		Integer pageNo = (Integer) queryParam.get("pageNo");
		Integer pageSize = (Integer) queryParam.get("pageSize");
//		Assert.notNull(query, "分页查询-查询条件不能为空");
//		Assert.notNull(pageNo, "分页查询-查询页码不能为空");
//		Assert.notNull(pageSize, "分页查询-查询每页数量不能为空");
		//查询总数
		long total = count(collection,query);
		//总共多少页,
		final Integer pages = (int) Math.ceil(total / (double) pageSize);
		if (pageNo <= 0) {
			pageNo = FIRST_PAGE_NUM;
		}
		int skip = pageSize * (pageNo - 1);
		FindIterable<Document> resultIterable = null;
		if (sort != null) {
			resultIterable = collection.find(query).sort(sort).skip(skip).limit(pageSize);
		} else {
			resultIterable = collection.find(query).skip(skip).limit(pageSize);
		}
		Map<String, Object> result = new HashMap<>();
		MongoCursor<Document> iterator = resultIterable.iterator();
		List<Document> list = new ArrayList<>();
		while(iterator.hasNext()){
			Document next = iterator.next();
			list.add(next);
		}
		result.put("rows", list);
		result.put("total", total);
		result.put("pages", pages);
		return result;
		
	}


	
	public static Double convertDecimal2Double(Document doc, String key) {
		Decimal128 decimal = doc.get(key, Decimal128.class);
		Double amount = SssfmNumberUtil.convertDecimal128toDouble(decimal);
		//将Decimal128 数据格式替换为double
		doc.replace(key, amount);
		return amount;
	}
	
	/**
	 * @Title: count
	 * @Description: 统计总数量
	 * @param collection
	 * @param query
	 * @return
	 */
	public static Long count(MongoCollection<Document> collection, Document query) {
		return collection.countDocuments(query);
	}

	/**
	 * 根据月份查询明细信息
	 * @param collection
	 * @param month
	 * @return
	 */
	public static List<Object> query(MongoCollection<Document> collection, int month) {
		List<Object> list  = collection.find(Filters.eq("month",month+"")).into(new ArrayList<Object>());
		return list;
	}
}

